package cn.doitedu.hbase.baseapi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;

/**
 * @date: 2019/7/9
 * @site: www.doitedu.cn
 * @author: hunter.d 涛哥
 * @qq: 657270652
 * @description: 表数据操作示例代码
 */
public class DMLDemos {


    Connection conn = null;

    @Before
    public void init() throws Exception {

        //Configuration conf = new Configuration();  它会自动加载classpath中的core-site.xml core-default.xml hdfs-site.xml hdfs-default.xml yarn-default.xml yarn-site.xml mapred-default.xml mapred-site.xml
        Configuration conf = HBaseConfiguration.create(); // 不光会加载hadoop的配置，还会加载  hbase-site.xml

        // 客户端连接不需要指定具体的master或者regionserver地址，只要指定zookeeper地址就行
        conf.set("hbase.zookeeper.quorum", "spark01:2181,spark02:2181,spark03:2181");

        // 本conn对象是一个lazy变量，真正使用时才生成
        conn = ConnectionFactory.createConnection(conf);

    }


    /**
     * 插入数据
     *
     * @throws Exception
     */
    @Test
    public void testPut() throws Exception {

        // 用conn获取一个表数据操作对象
        Table t1 = conn.getTable(TableName.valueOf("t1"));

        // 先构造一个数据封装对象
        Put r005 = new Put("r005".getBytes());
        r005.addColumn("base_info".getBytes(), "name".getBytes(), "崔莺莺".getBytes());
        r005.addColumn("base_info".getBytes(), "age".getBytes(), Bytes.toBytes("18"));
        r005.addColumn("extra_info".getBytes(), "gender".getBytes(), Bytes.toBytes("F"));

        Put r006 = new Put("r006".getBytes());
        r006.addColumn("base_info".getBytes(), "name".getBytes(), "西门庆".getBytes());
        r006.addColumn("base_info".getBytes(), "age".getBytes(), Bytes.toBytes("38"));
        r006.addColumn("extra_info".getBytes(), "gender".getBytes(), Bytes.toBytes("M"));


        ArrayList<Put> puts = new ArrayList<>();
        puts.add(r005);
        puts.add(r006);

        // 批次写入
        t1.put(puts);

        t1.close();
        conn.close();

    }

    /**
     * 比较mutator和table.put的差别
     * 1. mutator是一个异步操作，客户端先把数据写入本地的缓存，即返回，客户端不需要同步等待数据插入完成；而put是需要同步等待的
     * 2. mutator把数据放入本地缓存后，攒满一批再提交到hbase写入，这样可以提高数据插入的效率
     *
     * 对于静态批量数据（比如hdfs中已经存在的一堆文件）快速导入hbase，还有更加高效的办法：Bulkloader
     * 它的原理：  不需要通过网络RPC请求来提交数据，而是直接将原始文件，转换成hbase的底层文件HFILE，然后直接上传到hbase的表目录中
     *
     * @throws IOException
     */
    @Test
    public void testPut2() throws IOException {

        // 客户端如果有大量数据需要集中密集写入hbase表
        // 直接用put方法来做的话，客户端程序就需要等待put全部完成-- put是一个同步操作
        // 而BufferedMutator则允许客户端设置一个缓冲区，提交的数据先放在缓冲区，后面会异步提交到hbase集群
        BufferedMutator bufferedMutator = conn.getBufferedMutator(TableName.valueOf("t1".getBytes()));

        long start = System.currentTimeMillis();
        for(int i=0;i<100000;i++) {
            Put r = new Put(("z00"+i).getBytes());
            r.addColumn("base_info".getBytes(), "name".getBytes(), ("潘金莲"+i).getBytes());
            r.addColumn("base_info".getBytes(), "age".getBytes(), Bytes.toBytes(i+""));
            r.addColumn("extra_info".getBytes(), "gender".getBytes(), Bytes.toBytes("F"));

            // 这个动作是写入客户端的缓存，后续会按周期去提交到hbase集群
            bufferedMutator.mutate(r);
        }
        long end = System.currentTimeMillis();
        System.out.println(end-start);
        bufferedMutator.close();

        Table t1 = conn.getTable(TableName.valueOf("t1".getBytes()));

        start = System.currentTimeMillis();
        for(int i=0;i<100000;i++) {
            Put r = new Put(("p00"+i).getBytes());
            r.addColumn("base_info".getBytes(), "name".getBytes(), ("刘亦菲"+i).getBytes());
            r.addColumn("base_info".getBytes(), "age".getBytes(), Bytes.toBytes(i+""));
            r.addColumn("extra_info".getBytes(), "gender".getBytes(), Bytes.toBytes("F"));

            // 这个动作是写入客户端的缓存，后续会按周期去提交到hbase集群
            t1.put(r);
        }
        end = System.currentTimeMillis();
        System.out.println(end-start);


        t1.close();
        conn.close();
    }


    /**
     * 删除表中的 整行、或者某行的整列族，或者某行的某个列
     * @throws IOException
     */
    @Test
    public void testDeleteData() throws IOException {

        Table t1 = conn.getTable(TableName.valueOf("t1"));

        // delete参数对象中，如果只指定行键，则会删除整行的所有key-value
        Delete delete3 = new Delete("z003".getBytes());
        t1.delete(delete3);


        // delete参数对象中，制定了行键+列族，则会删除该行的指定列族中的key-value
        Delete delete4 = new Delete("z004".getBytes());
        delete4.addFamily("base_info".getBytes());
        t1.delete(delete4);


        // delete参数对象中，制定了行键+列族+列名，则会删除该列
        Delete delete5 = new Delete("z005".getBytes());
        delete5.addColumn("base_info".getBytes(),"name".getBytes());
        t1.delete(delete5);

        t1.close();
        conn.close();

    }


    /**
     * 清空整个表的数据，会保留表定义，还可以保留表的region划分
     * @throws IOException
     */
    @Test
    public void testTruncate() throws IOException {

        Admin admin = conn.getAdmin();

        admin.disableTable(TableName.valueOf("myspace:t3"));
        admin.truncateTable(TableName.valueOf("myspace:t3"),true);

        admin.close();
        conn.close();
    }




}
